/*
 * Copyright 2022 The Open Islands Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.openislands.oi.rpc;

import org.openislands.oi.constant.CommunicationProtocolEnum;
import org.openislands.oi.constant.Dict;
import org.openislands.oi.constant.RoleType;
import org.openislands.oi.constant.MessageCodeEnum;
import org.openislands.oi.error.RemoteCallException;
import org.openislands.oi.manager.CommonNodeManager;
import org.openislands.oi.scheduling.CoordinationResult;
import org.openislands.oi.error.LogicException;
import org.openislands.oi.pojo.dto.CommonResult;
import org.openislands.oi.pojo.dto.NodeDTO;
import org.openislands.oi.pojo.vo.CommonResponse;
import org.openislands.oi.rpc.base.*;
import org.openislands.oi.util.ConfiguratorUtils;
import org.openislands.oi.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
@SuppressWarnings("unused")
public class RPCExecutor implements ApplicationListener<ContextRefreshedEvent> {
    private static final Logger log = LoggerFactory.getLogger(RPCExecutor.class);

    @Resource
    private CommonNodeManager commonNodeManager;

    /**
     * Communication Protocol Executor Proxy
     * Load see {@link #onApplicationEvent(ContextRefreshedEvent)}
     * see <{@link CommunicationProtocolEnum#getType()}, {@link ExecutorProxy}>
     */
    private final Map<Class<?>, ExecutorProxy> executorProxyCache = new HashMap<>();

    /**
     * Simple rpc of the coordination many parties worker node
     *
     * @param nodes   many parties worker node
     * @param generic generic invoke target result
     * @param <T>     target type
     * @param <E>     result element type
     * @return many parties call result
     * @throws LogicException node info not found
     */
    public <T, E> CoordinationResult<E> simple(@NonNull Map<RoleType, List<String>> nodes,
                                               @NonNull Generic<T, CommonResponse<E>> generic) throws LogicException {
        CoordinationResult<E> coordinationResult = new CoordinationResult<>();
        Map<String, CommonResult<E>> dialedNodes = new HashMap<>();
        nodes.forEach((role, nodeIds) -> {
            nodeIds.forEach(nodeId -> {
                if (!dialedNodes.containsKey(nodeId)) {
                    CommonResult<E> oneOfResult = this.simple(nodeId, generic);
                    coordinationResult.addOneOfResult(role, nodeId, oneOfResult);
                    dialedNodes.put(nodeId, oneOfResult);
                } else {
                    coordinationResult.addOneOfResult(role, nodeId, dialedNodes.get(nodeId));
                }
            });
        });
        coordinationResult.calculateCoordinationResult();
        return coordinationResult;
    }

    /**
     * Simple rpc of the target worker node
     *
     * @param nodeId  target worker node id
     * @param generic generic invoke target result
     * @param <T>     target type
     * @param <E>     result element type
     * @return single node call result
     * @throws LogicException node info not found
     */
    public <T, E> CommonResult<E> simple(@NonNull String nodeId, @NonNull Generic<T, CommonResponse<E>> generic) throws LogicException {
        // local
        if (commonNodeManager.myNodeId(nodeId)) {
            return internal(Dict.WORKER_MARK, generic);
        }
        // get node strategy
        NodeDTO nodeDTO = commonNodeManager.queryByNodeId(nodeId);
        LogicException.nonNull(nodeDTO, MessageCodeEnum.DATA_NOT_FOUND_ERROR, String.format("can not found target node %s route info", nodeId));
        CommunicationProtocolEnum protocolEnum = CommunicationProtocolEnum.byName(nodeDTO.getCallStrategy());
        // proxy
        return genericInvoke(baseExtended -> {
            baseExtended.putItem(BaseExtended.ROUTE, ConfiguratorUtils.serviceRoute(Dict.FEDERATED_PROXY_MARK));
            baseExtended.putItem(BaseExtended.TARGET_MARK, nodeId);
            baseExtended.putItem(BaseExtended.SERVICE_MARK, Dict.WORKER_MARK);
        }, protocolEnum, generic);
    }

    /**
     * Service internal request, default use http protocol
     *
     * @param serviceMark target service name
     * @param generic     generic invoke target result
     * @param <T>         target type
     * @param <E>         result element type
     * @return call result
     */
    public <T, E> CommonResult<E> internal(@NonNull String serviceMark, @NonNull Generic<T, CommonResponse<E>> generic) {
        return genericInvoke(baseExtended -> {
            baseExtended.putItem(BaseExtended.ROUTE, ConfiguratorUtils.serviceRoute(serviceMark));
            baseExtended.putItem(BaseExtended.TARGET_MARK, StringUtils.EMPTY);
            baseExtended.putItem(BaseExtended.SERVICE_MARK, serviceMark);
        }, CommunicationProtocolEnum.HTTP, generic);
    }

    /**
     * The executor invocation process of the proxy generic target
     *
     * @param baseConsumer some setting for BaseExtended
     * @param protocolEnum communication protocol type enum
     * @param generic      generic target
     * @param <T>          target type
     * @param <E>          result element type
     * @return proxy generic target invoke result
     */
    private <T, E> CommonResult<E> genericInvoke(@NonNull Consumer<BaseExtended> baseConsumer,
                                                 @NonNull CommunicationProtocolEnum protocolEnum,
                                                 @NonNull Generic<T, CommonResponse<E>> generic) {
        try {
            // get executor proxy
            ExecutorProxy executorProxy = executorProxyCache.get(protocolEnum.getType());
            Executor<T> executor = executorProxy.getExecutor(generic.getParamsClass());
            // set extended config
            Extended<Class<?>, ItemExtended> extended = executor.extended();
            BaseExtended baseExtended = (BaseExtended) extended.get(BaseExtended.class, new BaseExtended());
            baseExtended.putItem(BaseExtended.RESPONSE_TYPE, generic.getResultType());
            baseConsumer.accept(baseExtended);
            // invoke executor
            CommonResponse<E> commonResponse = generic.invoke(executor.executor());
            return CommonResult.parse(commonResponse);
        } catch (RemoteCallException e) {
            return CommonResult.error(MessageCodeEnum.CALL_CONNECT_ERROR);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return CommonResult.error(MessageCodeEnum.CALL_CONNECT_ERROR);
        }
    }

    /**
     * Cache all protocol executor proxy
     *
     * @param event spring boot context refreshed
     */
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        ApplicationContext applicationContext = event.getApplicationContext();
        final Map<String, ExecutorProxy> beansOfType = applicationContext.getBeansOfType(ExecutorProxy.class);
        executorProxyCache.putAll(
                beansOfType.values().stream().collect(Collectors.toMap(ExecutorProxy::getClass, Function.identity()))
        );
    }
}
